package com.bw.gmall.realtime.app.dws;

import com.bw.gmall.realtime.bean.ShopNewVisitOrderBean;
import com.bw.gmall.realtime.utils.MyClickHouseUtil;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;


/*
*
* dws  1.  关连  把多个事实表连载一起       开窗        关联维度
*
* */
public class ShopNewVisitOrderApp {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.executeSql("CREATE TABLE table1_result (\n" +
                "    us STRING COMMENT '用户标识',\n" +
                "    shops STRING COMMENT '店铺标识',\n" +
                "    dt STRING COMMENT '日期',\n" +
                "    rt TIMESTAMP(3) COMMENT '访问时间',\n" +
                "    total_amount DECIMAL(10,2) COMMENT '总金额',\n" +
                "    WATERMARK FOR rt AS rt - INTERVAL '5' SECOND\n" +  // 添加水位线定义
                ")" + MyKafkaUtil.getKafkaDDL("Zb1_dwd_shop_new_visit_order","Zb1_dwd_shop_new_visit_order"));
//        tableEnv.sqlQuery("select * from table1_result").execute().print();
        Table resultTable = tableEnv.sqlQuery("WITH window_stats AS (\n" +
                "    SELECT\n" +
                "        shops,\n" +
                "        TUMBLE_START(rt, INTERVAL '1' HOUR) AS window_start,\n" +
                "        TUMBLE_END(rt, INTERVAL '1' HOUR) AS window_end,\n" +
                "        COUNT(DISTINCT us) AS total_visits,\n" +
                "        COUNT(DISTINCT CASE WHEN total_amount > 0 THEN us END) AS paid_visits,\n" +
                "        SUM(total_amount) AS window_total_amount\n" +
                "    FROM table1_result\n" +
                "    GROUP BY shops, TUMBLE(rt, INTERVAL '1' HOUR)\n" +
                "),\n" +
                "shop_totals AS (\n" +
                "    SELECT\n" +
                "        shops,\n" +
                "        SUM(window_total_amount) AS shop_total_amount\n" +
                "    FROM window_stats\n" +
                "    GROUP BY shops\n" +
                ")\n" +
                "SELECT\n" +
                "    w.shops AS shops,\n" +
                "    cast(w.window_start as string) AS windowStart,\n" +
                "     cast(w.window_end as string) AS windowEnd,\n" +
                "    w.total_visits AS totalVisits,\n" +
                "    w.paid_visits AS paidVisits,\n" +
                "    (w.total_visits - w.paid_visits) AS unpaidVisits,\n" +
                "    ROUND(w.paid_visits * 100.0 / NULLIF(w.total_visits, 0), 2) AS conversionRate,\n" +
                "    ROUND(w.window_total_amount * 100.0 / NULLIF(s.shop_total_amount, 0), 2) AS amountPercentage,\n" +
                "    ROUND(\n" +
                "        CASE WHEN w.paid_visits > 0 THEN w.window_total_amount / w.paid_visits ELSE 0 END, \n" +
                "        2\n" +
                "    ) AS averageOrderValue\n" +
                "FROM window_stats w\n" +
                "JOIN shop_totals s ON w.shops = s.shops");
//        resultTable.execute().print();

// 合并结果

        DataStream<Tuple2<Boolean, ShopNewVisitOrderBean>> tuple2DataStream = tableEnv.toRetractStream(resultTable, ShopNewVisitOrderBean.class);

        SingleOutputStreamOperator<ShopNewVisitOrderBean> map = tuple2DataStream.map(t -> t.f1);
        map.print(">>>>>>>");
////         将Table转换为DataStream
////         写入ClickHouse
        map.addSink(MyClickHouseUtil.getSinkFunction("insert into table shop_new_visit_order_stats_zb1 values(?,?,?,?,?,?,?,?,?)"));

        env.execute("ShopNewVisitOrderApp");
    }
}